/******************************************************************************
 * Copyright 2024 zhangcong. All Rights Reserved.
 *****************************************************************************/

#include "business/recordbusiness/storetask_manager.h"
#include <string.h>
#include <fstream>
#include <sstream>
#include "zmq.h"
#include "curl/curl.h"
#include "common/common_log.h"
#include "nlohmann/json.hpp"

using json = nlohmann::json;

#define RECV_BUF_SIZE 1024*4
#define SEND_BUF_SIZE 1024*4
#define CORE_ZMQ_FILE "/app/conf/core_info"
#define FILE_PATH_STR "/data/sensor/"

StoreTaskManager* StoreTaskManager::m_Instance = nullptr;

static int32_t searchFunc(std::string filename, std::string data,
                          std::string &result)
{
    std::string cmd = "grep " + data + " " + filename + "  | sed 's/^.*= //g'";
    FILE *fstream = nullptr;

    char buf[256] = {0};
    if (nullptr == (fstream = popen(cmd.c_str(), "r")))
    {
        Common_Log_e("open cmd fail");
        return -1;
    }

    if (nullptr == fgets(buf, sizeof(buf), fstream))
    {
        Common_Log_e("fgets buff fail");
        return -1;
    }
    result = buf;
    result = result.substr(0, result.length() - 1);

    Common_Log_e("data:%s, result:%s, leng:%d\n", data.c_str(), result.c_str(), result.length());

    return 0;
}

StoreTaskManager *StoreTaskManager::GetInstance()
{
    if (nullptr == m_Instance) {
        m_Instance = new StoreTaskManager();
    }
    return m_Instance;
}

int32_t StoreTaskManager::init()
{
    m_context = zmq_init(1);
    m_sub = zmq_socket(m_context, ZMQ_SUB);
    m_pub = zmq_socket(m_context, ZMQ_PUB);

    std::string ipaddr;
    std::string pubport;
    std::string subport;

    int32_t nret = searchFunc(CORE_ZMQ_FILE, "core_ip", ipaddr);
    if (nret < 0)
    {
        Common_Log_e("parse coreip fail");
        return -1;
    }

    nret = searchFunc(CORE_ZMQ_FILE, "zmq_data_sub_core_port", subport);
    if (nret < 0)
    {
        Common_Log_e("parse zmq_data_sub_core_port fail");
        return -1;
    }

    nret = searchFunc(CORE_ZMQ_FILE, "zmq_data_pub_core_port", pubport);
    if (nret < 0)
    {
        Common_Log_e("parse zmq_data_pub_core_port fail");
        return -1;
    }

    nret = searchFunc(CORE_ZMQ_FILE, "zmq_data_sub_core_topic", m_subtopic);
    if (nret < 0)
    {
        Common_Log_e("parse zmq_data_sub_core_topic fail");
        return -1;
    }

    nret = searchFunc(CORE_ZMQ_FILE, "zmq_data_pub_core_topic", m_pubtopic);
    if (nret < 0)
    {
        Common_Log_e("parse zmq_data_pub_core_topic fail");
        return -1;
    }

    nret = searchFunc(CORE_ZMQ_FILE, "data_uuid", m_dataUuid);
    if (nret < 0)
    {
        Common_Log_e("parse data_uuid fail");
        return -1;
    }

    nret = searchFunc(CORE_ZMQ_FILE, "core_uuid", m_coreUuid);
    if (nret < 0)
    {
        Common_Log_e("parse core_uuid fail");
        return -1;
    }

    std::string suburl = "tcp://" + ipaddr + ":" + subport;
    std::string puburl = "tcp://*:"+ pubport;

    zmq_connect(m_sub, suburl.c_str());
    zmq_bind(m_pub, puburl.c_str());
    zmq_setsockopt(m_sub, ZMQ_SUBSCRIBE, m_subtopic.c_str(), m_subtopic.length());

    sleep(3);
    m_storeTaskRunFlag = true;
    m_storeTaskThread.reset(new std::thread([this]
                                            { StoreTaskFunc(); }));

    return 0;
}

int32_t StoreTaskManager::proc()
{
    return 0;
}

void StoreTaskManager::PubUploadFileEndInfo(json &fileEndInfo)
{
    char sendbuf[SEND_BUF_SIZE] = {0};
    snprintf(sendbuf, SEND_BUF_SIZE, "%s|%d|%s|", m_pubtopic.c_str(), fileEndInfo.dump().length(), fileEndInfo.dump().c_str());
    Common_Log_e("sendbuf:%s sendlen:%d\n", sendbuf, strlen(sendbuf));
    int32_t nret = zmq_send(m_pub, sendbuf, strlen(sendbuf), 0);
    if (nret < 0)
    {
        Common_Log_e("storetask zmq send fail");
    }
}

void StoreTaskManager::ReqCurStoreTask()
{
    json reqtask;
    reqtask["cmd"] = "storeTaskCmd";
    reqtask["subcmd"] = "refreshStoreTask";
    reqtask["cid"] = 0;
    reqtask["uuid"] = m_dataUuid;
    reqtask["p_uuid"] = m_coreUuid;
    reqtask["type"] = "service";

    char sendbuf[SEND_BUF_SIZE] = {0};
    snprintf(sendbuf, SEND_BUF_SIZE, "%s|%d|%s|", m_pubtopic.c_str(), reqtask.dump().length(), reqtask.dump().c_str());
    Common_Log_e("sendbuf:%s sendlen:%d\n", sendbuf, strlen(sendbuf));
    int32_t nret = zmq_send(m_pub, sendbuf, strlen(sendbuf), 0);
    if (nret < 0)
    {
        Common_Log_e("storetask zmq send fail");
    }
}

void StoreTaskManager::StoreTaskFunc()
{
    ReqCurStoreTask();
    while (m_storeTaskRunFlag)
    {
        Common_Log_e("zmq start");
        char recvbuf[RECV_BUF_SIZE] = {0};
        memset(recvbuf, 0, sizeof(recvbuf));
        int32_t nret = zmq_recv(m_sub, recvbuf, RECV_BUF_SIZE, 0);
        if (nret < 0)
        {
            Common_Log_e("storetask zmq recv fail");
            continue;
        }

        Common_Log_e("recv data:%s\n", recvbuf);
        std::string str = recvbuf;
        size_t index_0 = str.find('|');
        size_t index_1 = str.find('|', index_0 + 1);
        size_t index_2 = str.find('|', index_1 + 1);

        std::string recvtopic = str.substr(0, index_0);
        std::string datalen = str.substr(index_0 + 1, index_1 - index_0 - 1);
        std::string recvdata = str.substr(index_1 + 1, index_2 - index_1 - 1);

        Common_Log_e("recvtopic:%s\nrecvdata:%s\ndatalen:%d\n", recvtopic.c_str(), recvdata.c_str(), std::atoi(datalen.c_str()));
        int32_t ret = StoreTaskProto::GetInstance()->DeSerialize(reinterpret_cast<const uint8_t *>(recvdata.c_str()), std::atoi(datalen.c_str()));
        if (ret < 0)
        {
            Common_Log_e("proto DeSerialize error!!!");
            continue;
        }

        json replyInfo;
        json storeTaskInfo;
        StoreTaskProto::GetInstance()->GetPlatCmd(storeTaskInfo);
        Common_Log_e("proto DeSerialize str:%s, %d, %s %d!!!", storeTaskInfo.dump().c_str(),
         !storeTaskInfo["cmd"].empty(), storeTaskInfo["cmd"].dump().c_str(), storeTaskInfo["cmd"] == "storeTaskCmd");
        replyInfo = storeTaskInfo;

        if (!storeTaskInfo["cmd"].empty() && storeTaskInfo["cmd"] == "storeTaskCmd")
        {
            std::string subcmd = storeTaskInfo["subcmd"];
            printf("subcmd:%s\n", subcmd.c_str());
            if (subcmd == "add")
            {
                StoreTaskInfo info;
                info.upload_data = false;
                info.upload_file = true;
                info.coreCpuID = storeTaskInfo["params"]["coreCpuId"].empty() ? "" : storeTaskInfo["params"]["coreCpuId"];
                info.storeCpuID = storeTaskInfo["params"]["storeCpuId"].empty() ? "" : storeTaskInfo["params"]["storeCpuId"];
                info.storeTaskName = storeTaskInfo["params"]["storeTaskName"].empty()? "" : storeTaskInfo["params"]["storeTaskName"];
                info.zmqIp = storeTaskInfo["params"]["zmqAddr"].empty()? "" : storeTaskInfo["params"]["zmqAddr"];
                info.mccAddress = storeTaskInfo["params"]["mcc_address"].empty()? "" : storeTaskInfo["params"]["mcc_address"];
                info.createTime = storeTaskInfo["params"]["createTime"].empty() ? "" :  storeTaskInfo["params"]["createTime"];
                info.startTime = storeTaskInfo["params"]["startTime"].empty()? "" : storeTaskInfo["params"]["startTime"] ;
                info.endTime = storeTaskInfo["params"]["endTime"].empty()? "" : storeTaskInfo["params"]["endTime"];
                info.minioEndPoint = storeTaskInfo["params"]["minioEndpoint"].empty()? "" : storeTaskInfo["params"]["minioEndpoint"];
                info.minioBucketName = storeTaskInfo["params"]["minioBucketName"].empty()? "" : storeTaskInfo["params"]["minioBucketName"];
                info.minioAccessKey = storeTaskInfo["params"]["minioAccessKey"].empty()? "" : storeTaskInfo["params"]["minioAccessKey"];
                info.minioSecretKey = storeTaskInfo["params"]["minioSecretKey"].empty()? "" : storeTaskInfo["params"]["minioSecretKey"];
                info.jsonInfo = storeTaskInfo.dump();

                // if (storeTaskInfo["params"]["sn"].size() != storeTaskInfo["params"]["ch_param"].size())
                // {
                //     Common_Log_e("sn array size is not equal to ch_param array size");
                //     return;
                // }
                for (int32_t index = 0; index < storeTaskInfo["params"]["sn"].size(); index++)
                {
                    info.sn.push_back(storeTaskInfo["params"]["sn"][index]);
                }
                for (int32_t index = 0; index < storeTaskInfo["params"]["ch_param"].size(); index++)
                {
                    info.chParam.push_back(storeTaskInfo["params"]["ch_param"][index]);
                }
                AddStoreTask(storeTaskInfo["params"]["storeTaskId"], info);
            }
            else if (subcmd == "start")
            {
                StartStoreTask(storeTaskInfo["params"]["storeTaskId"]);
            }
            else if (subcmd == "pause")
            {
                PauseStoreTask(storeTaskInfo["params"]["storeTaskId"]);
            }
            else if (subcmd == "delTask")
            {
                DeleteStoreTask(storeTaskInfo["params"]["storeTaskId"]);
            }
            else if (subcmd == "delFileOrDir")
            {
                DeleteStoreFile(storeTaskInfo["params"]["fileName"]);
            }
            else if (subcmd == "realTime")
            {
                StoreTaskInfo info;
                info.upload_data = true;
                info.upload_file = false;
                info.coreCpuID = storeTaskInfo["params"]["coreCpuId"].empty() ? "" : storeTaskInfo["params"]["coreCpuId"];
                info.storeCpuID = storeTaskInfo["params"]["storeCpuId"].empty() ? "" : storeTaskInfo["params"]["storeCpuId"];
                info.zmqIp = storeTaskInfo["params"]["zmqAddr"].empty()? "" : storeTaskInfo["params"]["zmqAddr"];
                info.mccAddress = storeTaskInfo["params"]["mccAddress"].empty()? "" : storeTaskInfo["params"]["mccAddress"];
                info.wsUrl = storeTaskInfo["params"]["webSocket"].empty()? "" : storeTaskInfo["params"]["webSocket"];

                for (int32_t index = 0; index < storeTaskInfo["params"]["sn"].size(); index++)
                {
                    info.sn.push_back(storeTaskInfo["params"]["sn"][index]);
                }
                for (int32_t index = 0; index < storeTaskInfo["params"]["ch_param"].size(); index++)
                {
                    info.chParam.push_back(storeTaskInfo["params"]["ch_param"][index]);
                }
                std::string taskname = info.wsUrl + "_" + info.zmqIp + "_" + info.mccAddress;
                AddStoreTask(taskname, info);
                StartStoreTask(taskname);
            }

            replyInfo.erase("params");
            replyInfo["data"] = storeTaskInfo["params"];
            replyInfo["code"] = 200;
            replyInfo["result"] = "success";

            char sendbuf[SEND_BUF_SIZE] = {0};
            snprintf(sendbuf, SEND_BUF_SIZE, "%s|%d|%s|", m_pubtopic.c_str(), replyInfo.dump().length(), replyInfo.dump().c_str());
            printf("sendbuf:%s sendlen:%d\n", sendbuf, strlen(sendbuf));
            nret = zmq_send(m_pub, sendbuf, strlen(sendbuf), 0);
            if (nret < 0)
            {
                Common_Log_e("storetask zmq send head fail");
                continue;
            }
        }
    }
}

void StoreTaskManager::AddStoreTask(const std::string &taskID, const StoreTaskInfo &taskInfo)
{
    if (m_StoreTaskArray.count(taskID) > 0) {
        Common_Log_e("taskID %s exist", taskID.c_str());
        return;
    }

    auto task = std::make_shared<StoreTask>(taskInfo);
    m_StoreTaskArray[taskID] = task;
    Common_Log_e("AddStoreTask success taskID:%s, m_StoreTaskArray size:%d\n", taskID.c_str(), m_StoreTaskArray.size());
}

void StoreTaskManager::StartStoreTask(const std::string &taskID)
{
    if (m_StoreTaskArray.count(taskID) <= 0) {
        Common_Log_e("taskID %s not exist", taskID.c_str());
        return;
    }

    m_StoreTaskArray[taskID]->StartTask();
    Common_Log_e("StartStoreTask success taskID:%s, m_StoreTaskArray size:%d\n", taskID.c_str(), m_StoreTaskArray.size());
}

void StoreTaskManager::PauseStoreTask(const std::string &taskID)
{
     if (m_StoreTaskArray.count(taskID) <= 0) {
        Common_Log_e("taskID %s not exist", taskID.c_str());
        return;
    }

   m_StoreTaskArray[taskID]->PauseTask();
   Common_Log_e("PauseStoreTask success taskID:%s, m_StoreTaskArray size:%d\n", taskID.c_str(), m_StoreTaskArray.size());
}

void StoreTaskManager::DeleteStoreTask(const std::string &taskID)
{
     if (m_StoreTaskArray.count(taskID) <= 0) {
        Common_Log_e("taskID %s not exist", taskID.c_str());
        return;
    }

    PauseStoreTask(taskID);
    m_StoreTaskArray.erase(taskID);
    Common_Log_e("DeleteStoreTask success taskID:%s, m_StoreTaskArray size:%d\n", taskID.c_str(), m_StoreTaskArray.size());
}

void StoreTaskManager::DeleteStoreFile(const std::string &filedir)
{
    char cmdline[256] = {0};
    sprintf(cmdline, "rm %s%s -rf", FILE_PATH_STR, filedir.c_str());
    system(cmdline);
    Common_Log_e("DeleteStoreFile success filedir:%s", filedir.c_str());
}
